package cn.llin.kafka;

import cn.llin.common.json.JSONObject;

import cn.llin.websocket.WebSocketUsers;
import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.Properties;




@Service
public class KafkaConsumerApplication {

//    private static final String TOPIC="topic_test_xh"; //"BA4035_BA4035010001_mhhwclxsgjb";//
    private static final String TOPIC="SHEHUIZHILI-TRACK-PERSON";
    private static final String CONFIG_DIR="D:\\MT\\赣州\\视频云对接20230217\\星环kafka对接使用说明\\";
//    private static final String CONFIG_DIR = "/dlxx_BWCW/rydlxx-test/";
    private static final String BOOTSTRAP_SERVERS="11.100.5.102:32610,11.100.5.101:32283,11.100.5.99:31994";
    private static final String TOS_PRINCIPAL="tos_k2142rz";


    private boolean stopMe = true;

    public void stopMe() {
        stopMe = false;
    }




//    private static   WebSocketServlet webSocketServlet;
//    @Autowired
//    public void  setDataVehicleWarningService(WebSocketServlet webSocketServlet){
//        KafkaConsumerApplication.webSocketServlet = webSocketServlet;
//    }



    @PostConstruct
    public void run() throws Exception{
        new Thread(new Runnable() {
            @Override
            public void run() {

            try {
                    Thread.sleep(14000);
                // 集群krb5.conf文件配置
                System.setProperty("java.security.krb5.conf", CONFIG_DIR + "krb5.conf");
                // 集群jaas.conf文件配置
                System.setProperty("java.security.auth.login.config", CONFIG_DIR + "jaas.conf");
//       System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");
//             System.setProperty("sun.security.krb5.debug", "true");
//
//       // 集群kafka bootstrap-server
                Properties props = new Properties();
                props.put("bootstrap.servers", BOOTSTRAP_SERVERS);

                props.put("sasl.kerberos.service.principal.instance", "tos_k2142rz");
//           //指定kafka消费者组 命名规则 账号名
                props.put("group.id", "default");
                props.put("enable.auto.commit", "false");
                props.put("auto.commit.interval.ms", "1000");// acks
//            props.put("auto.offset.reset", "earliest");// retries
                props.put("auto.offset.reset", "earliest");// retries
                props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
                props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

                props.put("security.protocol", "SASL_PLAINTEXT");// security.protocol
                props.put("sasl.mechanism", "GSSAPI");// sasl.mechanism
                props.put("sasl.kerberos.service.name", "kafka");// sasl.kerberos.service.name
                System.out.println("读取配置信息成功！");

                KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
                System.out.println("开始创建");
                consumer.subscribe(Arrays.asList(TOPIC));
                System.err.println("----------开始消费------------");
                JSONObject jsonObject = new JSONObject();

                while (true) {
                    ConsumerRecords<String, String> records = consumer.poll(10);
                    for (ConsumerRecord<String, String> record : records) {
                        System.out.println("offset =" + record.offset() + ",value =" + record.value());

                        jsonObject.put("jxh",record.value());
//                        webSocketServlet.sendMessage(JSON.toJSONString(jsonObject));
//                     WebSocketUsers.sendMessageToUsersByText(JSON.toJSONString(jsonObject));

//                        if (!stopMe){
////                        consumer.wakeup();
//                            consumer.close();
//                        }
                    }


                }
            } catch (Exception e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
           }
        }).start();

        }


//    }

    public static void main(String[] args) throws Exception {
        KafkaConsumerApplication kafkaConsumerApplication = new KafkaConsumerApplication();
        kafkaConsumerApplication.run();    //不能用这个启动
    }



}
